package org.qee.mq.custom;

import lombok.Data;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;

/**
 * @ProjectName: learning
 * @Package: org.qee.mq.custom
 * @ClassName: MappedFile
 * @Description:
 * @Date: 2021/12/20 2:18 下午
 * @Version: 1.0
 */
public class MappedFile {

    //public static int MAPPED_FILE_SIZE = 1024 * 1024 * 1024;

    public static int MAPPED_FILE_SIZE = 3 * 1024 * 1024;


    private FileChannel fileChannel;

    private MappedByteBuffer byteBuffer;

    private long currentWritePosition;

    // 目前已内存存储 每个消息的begin位置，可以实现为索引文件，目前简单实现
    private List<Long> msgBeginOffsetList = new ArrayList<>();

    private int indexFile;

    public MappedFile(String mappedFile) throws IOException {
        RandomAccessFile file = new RandomAccessFile(mappedFile, "rw");
        this.fileChannel = file.getChannel();
        this.byteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, MAPPED_FILE_SIZE);
        this.currentWritePosition = 0;
        indexFile = Integer.parseInt(mappedFile.substring(mappedFile.indexOf("-") + 1));
    }

    public WriteResult appendMsg(Message message) throws IOException {
        int len = message.getSize() + 4;
        if (currentWritePosition + len > MAPPED_FILE_SIZE) {
            return WriteResult.END_OF_FILE;
        }
        msgBeginOffsetList.add((long) byteBuffer.position());
        byteBuffer.putInt(message.getSize());
        byteBuffer.put(message.getBody());
        currentWritePosition += len;
        byteBuffer.force(); //简单先，每次都刷盘
        return WriteResult.OK;
    }

    public long getBeginOffset(long offset) {
        if (offset > msgBeginOffsetList.get(msgBeginOffsetList.size() - 1)) {
            return -1;
        }
        int left = 0;
        int right = msgBeginOffsetList.size() - 1;
        while (left < right) {
            int mid = left + (right - left) / 2;
            Long midValue = msgBeginOffsetList.get(mid);
            if (midValue == offset) {
                return offset;
            }
            if (midValue < offset) {
                left = mid + 1;
            } else {
                right = mid;
            }
        }
        return msgBeginOffsetList.get(right);
    }


    public ReadResult read(long beginOffset, int msgCount) throws IOException {
        ReadResult readResult = new ReadResult();
        beginOffset = getBeginOffset(beginOffset);
        readResult.setIndexFile(indexFile);
        if (beginOffset == -1) {
            readResult.setEndOffset(fileChannel.position());
            readResult.setSize(0);
            return readResult;
        }

        ByteBuffer byteBuffer = this.byteBuffer.slice();
        byteBuffer.position((int) beginOffset);
        ByteBuffer slice = byteBuffer.slice();
        int alreadyReadSize = 0;
        int maxReadSize = getMaxReadSize(slice);
        int cnt = 0;
        for (int i = 0; i < msgCount && alreadyReadSize + 4 < maxReadSize; i++) {
            int bodylen = slice.getInt();
            alreadyReadSize += 4;
            if (bodylen == 0) {
                readResult.getBytes().add(new byte[0]);
            }
            byte[] bytes = new byte[bodylen];
            slice.get(bytes);
            alreadyReadSize += bodylen;
            readResult.getBytes().add(bytes);
            cnt++;
        }
        readResult.setSize(cnt);
        readResult.setEndOffset(beginOffset + alreadyReadSize);
        return readResult;
    }

    private int getMaxReadSize(ByteBuffer byteBuffer) {
        return byteBuffer.limit();
    }

    public void recreateIndex() {
        msgBeginOffsetList.clear();
        ByteBuffer byteBuffer = this.byteBuffer.slice();
        int pos = 0;
        while (true) {
            byteBuffer.position(byteBuffer.position() + pos);
            int len = byteBuffer.getInt();
            if (len == 0) {
                break;
            }
            msgBeginOffsetList.add((long) (byteBuffer.position() - 4));
            pos = len;
        }
    }


    @Data
    public class ReadResult {

        private int size;

        private long endOffset;

        private Integer indexFile;

        private List<byte[]> result = new ArrayList<>();

        public List<byte[]> getBytes() {
            return result;
        }
    }

    public enum WriteResult {
        OK,
        END_OF_FILE
    }

}
